package com.chenbingkang.bigdata.homework3;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

public class Application {

    public static void main(String... args) {

        String tName = "chenbingkang:student";
        String cf1 = "info";
        String cf2 = "score";

        Configuration conf = HBaseConfiguration.create();
        //  conf.set("hbase.rootdir", "hdfs://47.101.206.249:9000/hbase");
        conf.set("hbase.zookeeper.quorum", "jikehadoop01:2181");

        Connection conn = null;
        Admin admin = null;
        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
            System.out.println(Arrays.toString(admin.listNamespaceDescriptors()));

            TableName tableName = TableName.valueOf(tName);
            if (admin.tableExists(tableName)) {
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }
            TableDescriptor descriptor = TableDescriptorBuilder.newBuilder(tableName)
                    .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(cf1.getBytes()).build())
                    .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(cf2.getBytes()).build())
                    .build();
            admin.createTable(descriptor);

        } catch (IOException e) {
            e.printStackTrace();
        }


        try {
            Table table = conn.getTable(TableName.valueOf(tName));
            Put put = new Put("Tom".getBytes());
            put.addColumn(cf1.getBytes(), "student_id".getBytes(), "20210000000001".getBytes());
            put.addColumn(cf1.getBytes(), "class".getBytes(), "1".getBytes());
            put.addColumn(cf2.getBytes(), "understanding".getBytes(), "75".getBytes());
            put.addColumn(cf2.getBytes(), "programming".getBytes(), "82".getBytes());
            table.put(put);

            put = new Put("Jerry".getBytes());
            put.addColumn(cf1.getBytes(), "student_id".getBytes(), "20210000000002".getBytes());
            put.addColumn(cf1.getBytes(), "class".getBytes(), "1".getBytes());
            put.addColumn(cf2.getBytes(), "understanding".getBytes(), "85".getBytes());
            put.addColumn(cf2.getBytes(), "programming".getBytes(), "67".getBytes());
            table.put(put);

            put = new Put("Jack".getBytes());
            put.addColumn(cf1.getBytes(), "student_id".getBytes(), "20210000000003".getBytes());
            put.addColumn(cf1.getBytes(), "class".getBytes(), "2".getBytes());
            put.addColumn(cf2.getBytes(), "understanding".getBytes(), "80".getBytes());
            put.addColumn(cf2.getBytes(), "programming".getBytes(), "80".getBytes());
            table.put(put);

            put = new Put("Rose".getBytes());
            put.addColumn(cf1.getBytes(), "student_id".getBytes(), "20210000000004".getBytes());
            put.addColumn(cf1.getBytes(), "class".getBytes(), "2".getBytes());
            put.addColumn(cf2.getBytes(), "understanding".getBytes(), "60".getBytes());
            put.addColumn(cf2.getBytes(), "programming".getBytes(), "61".getBytes());
            table.put(put);


            put = new Put("chenbingkang".getBytes());
            put.addColumn(cf1.getBytes(), "student_id".getBytes(), "20210000000005".getBytes());
            table.put(put);

            Scan scan = new Scan();
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                System.out.print(new String(result.getRow()) + " : ");
                Map<byte[], byte[]> cf1Map = result.getFamilyMap(cf1.getBytes());
                for (Map.Entry<byte[], byte[]> entry : cf1Map.entrySet()) {
                    System.out.print(new String(entry.getKey()) + "--" + new String(entry.getValue()) +"   ");
                }
                Map<byte[], byte[]> cf2Map = result.getFamilyMap(cf2.getBytes());
                for (Map.Entry<byte[], byte[]> entry : cf2Map.entrySet()) {
                    System.out.print(new String(entry.getKey()) + "--" + new String(entry.getValue()) +"   ");
                }
                System.out.println("");
            }
            Delete delete=new Delete("Tom".getBytes());
            table.delete(delete);

            System.out.println("=========================================");
            scan = new Scan();
            resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                System.out.print(new String(result.getRow()) + " : ");
                Map<byte[], byte[]> cf1Map = result.getFamilyMap(cf1.getBytes());
                for (Map.Entry<byte[], byte[]> entry : cf1Map.entrySet()) {
                    System.out.print(new String(entry.getKey()) + "--" + new String(entry.getValue()) +"   ");
                }
                Map<byte[], byte[]> cf2Map = result.getFamilyMap(cf2.getBytes());
                for (Map.Entry<byte[], byte[]> entry : cf2Map.entrySet()) {
                    System.out.print(new String(entry.getKey()) + "--" + new String(entry.getValue()) +"   ");
                }
                System.out.println("");
            }

            table.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
